#include "server_proxy.h"
#include "message.h"
#include "buffer.h"
#include "timewrap.h"
#include "Log.h"

int server_proxy::io_read_timeout = 5000;
int server_proxy::io_write_timeout = 1000;
message *recv_special_msg(int sock, message_builder_base *msg_builder)
{
	head msg_head;
	int ret = tcp_read_ms(sock, &msg_head, sizeof(msg_head), tcp_read_timeout);
	if (ret != sizeof(msg_head)) {
		return 0;
	}
	msg_head.ntoh();

	buffer buf;
	if (tcp_read_ms(sock, buf, msg_head.length, 5000) != msg_head.length) {
		LOG_ERROR_VA("message body receive failed, maybe bad format ");
		return 0;
	}
	return msg_builder->build_message(msg_head, buf);
}

int server_proxy_group::group_request(message &req_msg, message_builder_base &msg_builder, vector<response_message *> &results)
{
	vector<server_proxy *> alive_servers;
	for (vector<server_proxy>::iterator it = _servers.begin(); it != _servers.end(); it++) {
		if ((*it).check())
			alive_servers.push_back(&(*it));
	}
	if (alive_servers.empty()) {
		LOG_ERROR_VA("no alived match node");
		return 0;
	}

	int num_reqs = 0;
	int num_alives = alive_servers.size();
	int index = 0;
	for (int i = 0; i != num_alives; i++) {
		if (alive_servers[index]->send_msg(req_msg) <= 0) {
			LOG_ERROR_VA("group server <%s,%d> send msg error, server reset", 
			alive_servers[index]->get_address().ip().c_str(), alive_servers[index]->get_address().port());
			alive_servers[index]->reset();
			alive_servers.erase(alive_servers.begin() + index);
			index--;
		} else 
			num_reqs++;
		index++;
	}


	if (num_reqs == 0) {
		LOG_ERROR_VA("no mach request send to match nodes");
		return 0;
	}

	timee t = timee::now();
	int response_num = 0;
	int num_resps = alive_servers.size();
	while (response_num < num_resps) {
		if (timee::now() - t > 10 * 1000000) {
			for (int i = 0; i != alive_servers.size(); i++) 
				alive_servers[i]->reset();
			break;
		}
		
		pollfd fds[100];
		
		for (int i = 0; i != alive_servers.size(); i++) {
			fds[i].fd = alive_servers[i]->get_sock();
			fds[i].events = POLLIN;
		}
		int n = poll(fds, alive_servers.size(), 1000);
		if (n == -1) {
			LOG_ERROR_VA("poll() call in group reequest error");
			break;
		}

		int index = 0;
		num_alives = alive_servers.size();
		for (int i = 0; i != num_alives; i++) {
			if (fds[i].revents & POLLIN) {
				message *msg = recv_special_msg(fds[i].fd, &msg_builder);
				if (msg) {
					results.push_back(static_cast<response_message *>(msg));
				} else {
					alive_servers[index]->reset();
				}
				alive_servers.erase(alive_servers.begin() + index);
				index--;
				response_num++;
			}
			index++;
		}
	}
	return results.size();
}

